package com.reactim.message.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.reactim.message.connection.ConnectionDetails;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import reactor.core.publisher.Mono;

import java.time.Duration;

/**
 * Redis配置类
 * 配置Redis连接和序列化器
 */
@Configuration
@EnableConfigurationProperties(RedisProperties.class)
public class RedisConfig {

    @Value("${spring.application.name:reactim-message-service}")
    private String applicationName;

    /**
     * 配置用于ConnectionDetails的ReactiveRedisTemplate
     */
    @Bean
    public ReactiveRedisTemplate<String, ConnectionDetails> connectionDetailsRedisTemplate(
            ReactiveRedisConnectionFactory factory) {
        
        StringRedisSerializer keySerializer = new StringRedisSerializer();
        
        // 配置Jackson序列化器，支持Java 8日期时间类型
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        
        Jackson2JsonRedisSerializer<ConnectionDetails> valueSerializer = 
                new Jackson2JsonRedisSerializer<>(ConnectionDetails.class);
        valueSerializer.setObjectMapper(objectMapper);
        
        RedisSerializationContext.RedisSerializationContextBuilder<String, ConnectionDetails> builder =
                RedisSerializationContext.newSerializationContext(keySerializer);
        
        RedisSerializationContext<String, ConnectionDetails> context = builder
                .value(valueSerializer)
                .hashKey(keySerializer)
                .hashValue(valueSerializer)
                .build();
        
        return new ReactiveRedisTemplate<>(factory, context);
    }
    
    /**
     * 配置通用的ReactiveStringRedisTemplate
     */
    @Bean
    public ReactiveStringRedisTemplate reactiveStringRedisTemplate(
            ReactiveRedisConnectionFactory factory) {
        return new ReactiveStringRedisTemplate(factory);
    }

    /**
     * 配置通用的ReactiveRedisTemplate
     */
    @Bean
    public ReactiveRedisTemplate<String, Object> reactiveRedisTemplate(
            ReactiveRedisConnectionFactory factory) {
        
        Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<>(Object.class);
        
        RedisSerializationContext<String, Object> context = RedisSerializationContext
                .<String, Object>newSerializationContext()
                .key(RedisSerializationContext.SerializationPair.fromSerializer(RedisSerializer.string()))
                .value(RedisSerializationContext.SerializationPair.fromSerializer(serializer))
                .hashKey(RedisSerializationContext.SerializationPair.fromSerializer(RedisSerializer.string()))
                .hashValue(RedisSerializationContext.SerializationPair.fromSerializer(serializer))
                .build();
        
        return new ReactiveRedisTemplate<>(factory, context);
    }
    
    /**
     * 配置Redis连接工厂，支持连接池
     */
    @Bean
    @ConditionalOnProperty(name = "spring.redis.use-custom-connection", havingValue = "true")
    public ReactiveRedisConnectionFactory reactiveRedisConnectionFactory(RedisProperties redisProperties) {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(redisProperties.getHost());
        configuration.setPort(redisProperties.getPort());
        configuration.setPassword(redisProperties.getPassword());
        configuration.setDatabase(redisProperties.getDatabase());
        
        // 配置连接池
        GenericObjectPoolConfig<?> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(redisProperties.getLettuce().getPool().getMaxActive());
        poolConfig.setMaxIdle(redisProperties.getLettuce().getPool().getMaxIdle());
        poolConfig.setMinIdle(redisProperties.getLettuce().getPool().getMinIdle());
        poolConfig.setMaxWait(redisProperties.getLettuce().getPool().getMaxWait());
        
        LettucePoolingClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder()
                .commandTimeout(redisProperties.getTimeout())
                .poolConfig(poolConfig)
                .build();
        
        return new LettuceConnectionFactory(configuration, clientConfig);
    }
    
    /**
     * Redis健康检查指示器
     * 用于监控Redis连接状态
     */
    @Bean
    public HealthIndicator redisHealthIndicator(ReactiveRedisConnectionFactory redisConnectionFactory) {
        return () -> {
            ReactiveRedisConnection connection = null;
            try {
                connection = redisConnectionFactory.getReactiveConnection();
                Mono<String> ping = connection.ping();
                ping.block(Duration.ofSeconds(2));
                return Health.up().withDetail("version", connection.serverCommands().info().block()).build();
            } catch (Exception e) {
                return Health.down().withDetail("error", e.getMessage()).build();
            } finally {
                if (connection != null) {
                    connection.close();
                }
            }
        };
    }
    

}